package com.wwf;

import com.wwf.util.RpcRequest;
import com.wwf.util.ZookeeperClient;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;

import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;

/**
 * @author wwf
 * @date 2021年01月07日 14:38:38
 */
public class ClientBoostrap {
    private static String zkServers="103.45.186.39:2181,103.45.186.39:2182,103.45.186.39:2183";
    private static int connectionTimeout = 100000;
    private static Map<String,RPCConumer> rpcConumerMap = new ConcurrentHashMap<String, RPCConumer>();
   // private static String providerName="UserService#sayHello#";
    public static void main(String[] args) throws InterruptedException {
        //初始化消费端实例
        ZkClient zkClient = initRpcConumers();
        //启动定时刷新节点响应时间
        refeshRespsTime(zkClient);
        for (;;){
            //获取负载的消费端实例
            RPCConumer rpcConumer = getLoadsConumer(zkClient);
            if(rpcConumer != null){
                IUserService userService = (IUserService)rpcConumer.creatProxy(IUserService.class,new RpcRequest());
                long starttimes = System.currentTimeMillis()/1000;
                String s = userService.sayHello("hello server");
                long endtimes = System.currentTimeMillis();
                long resptime = endtimes-starttimes;
                String hostnode = ZookeeperClient.getNode()+"/"+rpcConumer.getHostnode();
                long createRespNodeTimes = System.currentTimeMillis();
                String hostRspNodeValue = createRespNodeTimes+"-"+resptime;//记录节点响应时间格式：记录时间-耗时
                if(zkClient.exists(hostnode)){
                    zkClient.writeData(hostnode,hostRspNodeValue);
                }
                System.out.println("服务端："+rpcConumer.getIp()+":"+rpcConumer.getPort()+"返回值："+s);
            }
            Thread.sleep(10);
        }
    }

    /**
     * 定时刷新节点的响应时间
     * @param zkClient
     */
    public static void refeshRespsTime(final ZkClient zkClient){
        new Thread(new Runnable() {
            public void run() {
                while (true){
                    try {
                        Iterator<Map.Entry<String, RPCConumer>> iterator = rpcConumerMap.entrySet().iterator();
                        while (iterator.hasNext()){
                            Map.Entry<String, RPCConumer> next = iterator.next();
                            RPCConumer value = next.getValue();
                            String hostnode = ZookeeperClient.getNode()+"/"+value.getHostnode();
                            if(zkClient.exists(hostnode)){
                                String respCtime = zkClient.readData(hostnode);
                                if(respCtime != null &&  !"".equals(respCtime)){
                                    try {
                                        Long createTimes = Long.parseLong(respCtime.substring(0, respCtime.indexOf("-")));
                                        Long currentTimes = System.currentTimeMillis();
                                        Long difTimes = currentTimes-createTimes;
                                        if(difTimes != null &&difTimes >5000){//超过五秒未更新需要清空节点响应时间记录
                                            zkClient.writeData(hostnode,"");
                                        }
                                    }catch (Exception e){
                                        e.printStackTrace();
                                    }

                                }
                            }
                        }
                        System.out.println("定时刷新节点执行时间");
                        Thread.sleep(5000);//五秒一次
                    } catch (InterruptedException exception) {
                        exception.printStackTrace();
                    }
                }
            }
        }).start();
    }

    /**
     * 获取负载的消费端实例(内写负载均衡策略)
     * @return
     */
    protected static RPCConumer getLoadsConumer(ZkClient zkClient){
        RPCConumer rpcConumer = null;
        long minRepspTime=0;
        Iterator<Map.Entry<String, RPCConumer>> iterator = rpcConumerMap.entrySet().iterator();
        /**
         * 按服务提供者响应时间来选取负载
         */
        
        while (iterator.hasNext()){
            Map.Entry<String, RPCConumer> mapV = iterator.next();
            RPCConumer value = mapV.getValue();
            String hostnode = value.getHostnode();
            String hostRespNode = ZookeeperClient.getNode()+"/"+hostnode;
            if(zkClient.exists(hostRespNode)){
                String respCtime = zkClient.readData(hostRespNode);
                if(respCtime != null &&  !"".equals(respCtime)){
                    try {
                        long curRepspTime = Long.parseLong(respCtime.substring(respCtime.indexOf("-")+1,respCtime.length()));
                        if(minRepspTime==0||curRepspTime<minRepspTime){
                            minRepspTime = curRepspTime;
                            rpcConumer = value;
                        }
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }else{
                    rpcConumer = value;
                    break;
                }
            }
        }
        /**
         * 随机负载
         */
        if(rpcConumer == null){//都没有响应时间需要随机负载
            int size = rpcConumerMap.entrySet().size();
            int keyindex = 0;
            int countkey=0;
            if(size >0){
                keyindex =  new Random().nextInt(size);
                if(keyindex == size) keyindex = keyindex-1;
            }
            Iterator<Map.Entry<String, RPCConumer>> iterator1 = rpcConumerMap.entrySet().iterator();
            while (iterator1.hasNext()){
                Map.Entry<String, RPCConumer> next = iterator1.next();
                RPCConumer currentRpcConumer = next.getValue();
                if(countkey == keyindex){
                    rpcConumer = next.getValue();
                }
                countkey++;
                if(rpcConumer != null) break;
            }
        }

        return rpcConumer;
    }
    /**
     * 初始化消费端实例
     */
    protected static ZkClient initRpcConumers(){
        //链接zookeeper获取服务端地址列表
        ZookeeperClient zookeeperClient = new ZookeeperClient(zkServers, connectionTimeout);
        final ZkClient zkClient = zookeeperClient.getZkClient();
        final String path = zookeeperClient.getNode();
        List<String> children = zkClient.getChildren(path);
        for (String child : children) {
           // String hostname=zkClient.readData(path+"/"+child);
            doInitRpcConumers(child);
        }
        //订阅服务器节点变更通知
        zkClient.subscribeChildChanges(path, new IZkChildListener() {
            public void handleChildChange(String parentPath, List<String> children) throws Exception {
                updateRpcConumerMap(children);
            }
        });
        return zkClient;
    }

    /**
     * 真正实例化RpcConumers的方法
     */
    protected static void doInitRpcConumers(String hostname){
        if(hostname != null && !"".equals(hostname)){
            int splitindex = hostname.indexOf(":");
            String ip = hostname.substring(0, splitindex);
            String portStr = hostname.substring(splitindex+1,hostname.length());
            int port = Integer.parseInt(portStr);
            RPCConumer rpcConumer = new RPCConumer(ip,port);
            rpcConumer.setHostnode(hostname);
            rpcConumerMap.put(hostname,rpcConumer);
            //bootstrap.connect(ip,port).sync();
            System.out.println("客户端连接到服务端地址为："+hostname);
        }
    }

    /**
     * 跟新消费端连接服务器的实例化信息
     * @param children
     */
    protected static void updateRpcConumerMap(List<String> children){
       
        Iterator<Map.Entry<String, RPCConumer>> iterator = rpcConumerMap.entrySet().iterator();
        while (iterator.hasNext()){
            Map.Entry<String, RPCConumer> next = iterator.next();
            String key = next.getKey();
            RPCConumer value = next.getValue();
            boolean mapConsumerIsLive=false;
            for (String child : children) {
                String hostname=child;
                if(key.equals(hostname)){//说明节点还存活
                    mapConsumerIsLive=true;
                }
                if(rpcConumerMap.get(hostname)==null){//新的节点加入需要更新rpcConumerMap
                    doInitRpcConumers(hostname);
                }
            }
            if(!mapConsumerIsLive){//没有存活
                ExecutorService service = value.getService();
                service.shutdown();
                iterator.remove();//移除服务节点
            }
        }
    }
}
